package com.atguigu.actual.edu0417.app.dwd.db;

import com.atguigu.actual.edu0417.app.BaseApp;
import com.atguigu.actual.edu0417.app.SqlBaseApp;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: 洛尘
 * @since: 2023-10-20 14:50
 * @description: 加购行为
 **/
public class DwdTradeCartAdd extends SqlBaseApp {
    public static void main(String[] args) {
        DwdTradeCartAdd dwdTradeCartAdd = new DwdTradeCartAdd();
        dwdTradeCartAdd.base(args);
    }


    @Override
    public void process(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_cart_add_group"));

        //过滤加购数据
        Table cartTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['cart_price'] cart_price,\n" +
                "`data`['deleted'] deleted,\n" +
                "`data`['sold'] sold,\n" +
                "ts\n"+
                "from\n" +
                "topic_db where `table`='cart_info' \n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("cart_add",cartTable);
//        tableEnv.executeSql("select * from cart_add").print();

        //将过滤出来的数据写入kafka主题
        tableEnv.executeSql("create table dwd_trade_cart_add\n" +
                "(id string,\n" +
                "user_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "cart_price string,\n" +
                "deleted string,\n" +
                "sold string,\n" +
                "ts string,\n" +
                "primary key (id) NOT ENFORCED" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cart_add"));
        //写入
//        tableEnv.executeSql("select * from dwd_trade_cart_add").print();
        tableEnv.executeSql("insert into dwd_trade_cart_add select * from cart_add");
    }
}